package com.chapter.sample.toolkit.lock;

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.MeterRegistry;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.util.Assert;

import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import jakarta.validation.constraints.NotNull;

/**
 * Redis 分布式锁执行器
 *
 * <p>Example:</p>
 * <pre>{@code
 * @Autowired
 * private RedisLockExecutor executor;
 *
 * try {
 *     executor.lock("lock:order:t0001");
 *     // TODO Logic
 * } finally {
 *     executor.unlock(key);
 * }
 *
 * if (executor.tryLock("lock:order:t0002")) {
 *     try {
 *          // TODO Logic
 *     } finally {
 *          executor.unlock(key);
 *     }
 * }
 *
 * if (executor.tryLock("lock:order:t0003", Duration.ofSeconds(1L))) {
 *     try {
 *          // TODO Logic
 *     } finally {
 *          executor.unlock(key);
 *     }
 * }
 *
 * // 无返回值
 * executor.executeWithLock("lock:order:t0004", () -> {
 *    // TODO Logic
 * });
 *
 * // 有返回值
 * final Optional<Order> opt = executor.executeWithTryLock("lock:order:t0005", () -> {
 *    // TODO Logic
 *    return order;
 * });
 *
 * // 无返回值
 * executor.executeWithTryLock("lock:order:t0006", () -> {
 *    // TODO Logic
 * });
 *
 * // 有返回值
 * final Optional<Order> opt = executor.executeWithTryLock("lock:order:t0007", () -> {
 *    // TODO Logic
 *    return order;
 * });
 *
 * }
 * </pre>
 *
 * @author zlikun
 * @created 2022/1/15 15:30
 * @see com.chapter.sample.toolkit.cache.config.RedissonAutoConfiguration
 */
@Slf4j
public class RedisLockExecutor implements LockExecutor {

    // Redisson客户端
    private RedissonClient client;
    // Prometheus计数器
    private MeterRegistry registry;

    private Counter lockedCounter;
    private Counter unlockCounter;
    private Counter tryLockTrueCounter;
    private Counter tryLockFalseCounter;

    public RedisLockExecutor(RedissonClient client, MeterRegistry registry) {
        Assert.notNull(client, "RedissonClient 参数不能为空");
        Assert.notNull(registry, "MeterRegistry 参数不能为空");
        this.client = client;
        this.registry = registry;

        lockedCounter = Counter.builder("game.redisson.lock_use")
                .tag("action", "locked")
                .description("Number of redisson lock")
                .register(registry);

        unlockCounter = Counter.builder("game.redisson.lock_use")
                .tag("action", "unlocked")
                .register(registry);

        tryLockTrueCounter = Counter.builder("game.redisson.lock_use")
                .tag("action", "try_lock_true")
                .register(registry);

        tryLockFalseCounter = Counter.builder("game.redisson.lock_use")
                .tag("action", "try_lock_false")
                .register(registry);

    }

    private RLock getLock(@NotNull String key) {
        return client.getLock(String.format("r_lock:%s", key));
    }

    /**
     * 加锁，默认30s超时，每10s会自动续期30s，必须与 #unlock(String) 结对使用
     *
     * @param key
     * @see #executeWithLock(String, Runnable)
     * @see #executeWithLock(String, Callable)
     */
    @Override
    public void lock(@NotNull String key) {
        getLock(key).lock();
        lockedCounter.increment();
    }

    /**
     * 不建议使用，尽快更换
     *
     * @param key
     * @param timeout
     * @see #lock(String)
     */
    @Deprecated
    //@GameDeprecated(since = "2022/01/15")
    public void lock(@NotNull String key, int timeout) {
        lock(key, TimeUnit.SECONDS, timeout);
    }

    /**
     * 不建议使用，尽快更换
     *
     * @param key
     * @param unit
     * @param timeout
     * @see #lock(String)
     */
    @Deprecated
    //@GameDeprecated(since = "2022/01/15")
    public void lock(@NotNull String key, @NotNull TimeUnit unit, int timeout) {
        getLock(key).lock(timeout, unit);
        lockedCounter.increment();
    }

    /**
     * 释放锁
     *
     * @param key
     */
    @Override
    public void unlock(@NotNull String key) {
        unlock(getLock(key));
    }

    /**
     * 释放锁
     *
     * @param lock
     */
    private void unlock(RLock lock) {
        try {
            lock.unlock();
            unlockCounter.increment();
        } catch (Exception e) {
            log.error("分布式锁解锁失败 key: {}", lock.getName(), e);
        }
    }

    /**
     * 尝试获取锁（不等待），取不到返回 false，取到则返回 true，超时时间与续期机制与 #lock(String) 一致，必须与 #unlock(String) 结对使用
     *
     * @param key
     * @return
     * @see #executeWithLock(String, Runnable)
     * @see #executeWithLock(String, Callable)
     */
    @Override
    public boolean tryLock(@NotNull String key) {
        boolean flag = getLock(key).tryLock();
        if (flag) {
            tryLockTrueCounter.increment();
        } else {
            tryLockFalseCounter.increment();
        }
        return flag;
    }

    /**
     * 尝试获取锁（等待 wait 时间），取不到返回 false，取到则返回 true，超时时间与续期机制与 #lock(String) 一致，必须与 #unlock(String) 结对使用
     *
     * @param key
     * @param waitTime
     * @return
     * @throws InterruptedException
     * @see #executeWithTryLock(String, Duration, Runnable)
     * @see #executeWithTryLock(String, Duration, Callable)
     */
    @Override
    public boolean tryLock(@NotNull String key, @NotNull Duration waitTime) throws InterruptedException {
        boolean flag = getLock(key).tryLock(waitTime.toMillis(), TimeUnit.MILLISECONDS);
        if (flag) {
            tryLockTrueCounter.increment();
        } else {
            tryLockFalseCounter.increment();
        }
        return flag;
    }

    /**
     * 不建议使用的锁，请尽快更换
     *
     * @param key
     * @param waitTime
     * @param leaseTime
     * @param unit
     * @return
     * @throws InterruptedException
     * @see #tryLock(String, Duration)
     * @see #tryLock(String)
     */
    @Deprecated
    //@GameDeprecated(since = "2022/01/15")
    public boolean tryLock(@NotNull String key, long waitTime, long leaseTime, @NotNull TimeUnit unit) throws InterruptedException {
        final boolean flag = getLock(key).tryLock(waitTime, leaseTime, unit);
        if (flag) {
            tryLockTrueCounter.increment();
        } else {
            tryLockFalseCounter.increment();
        }
        return flag;
    }

    /**
     * 判断是否加锁
     *
     * @param key
     * @return
     */
    @Override
    public boolean isLocked(@NotNull String key) {
        return getLock(key).isLocked();
    }

    /**
     * 判断锁是否被当前线程持有
     *
     * @param key
     * @return
     */
    public boolean isHeldByCurrentThread(@NotNull String key) {
        return getLock(key).isHeldByCurrentThread();
    }

    /**
     * 带分布式锁执行器
     *
     * @param key
     * @param callable
     * @param <T>
     * @return
     * @throws Exception
     */
    @Override
    public <T> Optional<T> executeWithLock(@NotNull String key, @NotNull Callable<T> callable) throws Exception {
        final RLock lock = getLock(key);
        try {
            lock.lock();
            lockedCounter.increment();
            return Optional.ofNullable(callable.call());
        } finally {
            unlock(lock);
        }
    }

    /**
     * 带分布式锁执行器
     *
     * @param key
     * @param runnable
     * @throws Exception
     */
    @Override
    public void executeWithLock(@NotNull String key, @NotNull Runnable runnable) throws Exception {
        final RLock lock = getLock(key);
        try {
            lock.lock();
            lockedCounter.increment();
            runnable.run();
        } finally {
            unlock(lock);
        }
    }

    /**
     * 执行器带TryLock机制（自动续租）
     *
     * @param key
     * @param callable
     * @param <T>
     * @return
     * @throws Exception
     */
    @Override
    public <T> Optional<T> executeWithTryLock(@NotNull String key, @NotNull Callable<T> callable) throws Exception {
        final RLock lock = getLock(key);
        if (lock.tryLock()) {
            try {
                tryLockTrueCounter.increment();
                return Optional.ofNullable(callable.call());
            } finally {
                unlock(lock);
            }
        } else {
            tryLockFalseCounter.increment();
            return Optional.empty();
        }
    }

    /**
     * 执行器带TryLock机制（自动续租）
     *
     * @param key
     * @param runnable
     * @throws Exception
     */
    @Override
    public void executeWithTryLock(@NotNull String key, @NotNull Runnable runnable) throws Exception {
        final RLock lock = getLock(key);
        if (lock.tryLock()) {
            try {
                tryLockTrueCounter.increment();
                runnable.run();
            } finally {
                unlock(lock);
            }
        } else {
            tryLockFalseCounter.increment();
        }
    }

    /**
     * 执行器带TryLock机制（自动续租）
     *
     * @param key
     * @param callable
     * @param <T>
     * @return
     * @throws Exception
     */
    @Override
    public <T> Optional<T> executeWithTryLock(@NotNull String key, @NotNull Duration waitTime, @NotNull Callable<T> callable) throws Exception {
        final RLock lock = getLock(key);
        if (lock.tryLock(waitTime.toMillis(), TimeUnit.MILLISECONDS)) {
            try {
                tryLockTrueCounter.increment();
                return Optional.ofNullable(callable.call());
            } finally {
                unlock(lock);
            }
        } else {
            tryLockFalseCounter.increment();
            return Optional.empty();
        }
    }

    /**
     * 执行器带TryLock机制（自动续租）
     *
     * @param key
     * @param waitTime
     * @param runnable
     * @throws Exception
     */
    @Override
    public void executeWithTryLock(@NotNull String key, @NotNull Duration waitTime, @NotNull Runnable runnable) throws Exception {
        final RLock lock = getLock(key);
        if (lock.tryLock(waitTime.toMillis(), TimeUnit.MILLISECONDS)) {
            try {
                tryLockTrueCounter.increment();
                runnable.run();
            } finally {
                unlock(lock);
            }
        } else {
            tryLockFalseCounter.increment();
        }
    }

}
